package com.bridgeintelligent.tag.bulkload.service.core.threadpool;

import com.bridgeintelligent.tag.bulkload.api.BulkLoadConfig;
import com.bridgeintelligent.tag.bulkload.api.BulkLoadException;
import com.wayneleo.quickstart.framework.ExceptionCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * Created by JackCheung on 2019/9/19 8:26 PM.
 */
@Slf4j
@Component
public class ThreadPoolExecutor {

    private Executor executor;

    public ThreadPoolExecutor(@Qualifier("threadPool_realTime") Executor executor) {
        this.executor = executor;
    }

    /**
     * 使用多线程处理任务
     *
     * @param indexName
     * @param list
     * @param delList
     * @param type
     */
    public void handler(String indexName, String indexType, List<Map<String, Map<String, Object>>> list, List<Map<String, String>> delList, int type) {
        if (CollectionUtils.isNotEmpty(list) || CollectionUtils.isNotEmpty(delList)) {
            CountDownLatch countDownLatch = new CountDownLatch(BulkLoadConfig.THREAD_COUNT);
            if (CollectionUtils.isNotEmpty(list)) {
                //修改任务
                list.forEach(map -> {
                    RealTimeThread realTimeThread = new RealTimeThread(countDownLatch, indexName, map, null, type, indexType);
                    executor.execute(realTimeThread);
                });
            } else {
                //删除任务
                delList.forEach(map -> {
                    RealTimeThread realTimeThread = new RealTimeThread(countDownLatch, indexName, null, map, type, indexType);
                    executor.execute(realTimeThread);
                });
            }
            try {
                long timeout = 3600L;
                countDownLatch.await(timeout, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
                log.error("线程同步发生异常->{}", e.getMessage());
                Thread.currentThread()
                      .interrupt();
                throw new BulkLoadException(new ExceptionCode(999, e.getMessage()));
            }
        }
    }
}
